package com.tang.framework.delay.support;

import com.tang.common.properties.sys.CoolTeaProperties;
import com.tang.framework.delay.DelayQueue;
import com.tang.framework.delay.support.domain.DelayMethod;
import com.tang.framework.delay.support.domain.Message;
import jodd.util.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.TransactionDefinition;

import jakarta.annotation.Resource;

import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.*;
import java.util.concurrent.*;


@Slf4j
public class DelayExecute {

    private final ConcurrentHashMap<String, List<DelayMethodHandler>> delayMethodMap = new ConcurrentHashMap<>(256);

    private final LinkedList<String> queueNameList = new LinkedList<>();

    private final ConcurrentHashMap<String, RDelayedQueue<Message<?>>> delayedQueueMap = new ConcurrentHashMap<>(256);

    private final ConcurrentHashMap<String, RBlockingQueue<Message<?>>> blockingQueueMap = new ConcurrentHashMap<>(256);

    private final RedissonClient redissonClient;


    private DataSourceTransactionManager dataSourceTransactionManager;


    private TransactionDefinition transactionDefinition;

    @Resource
    private CoolTeaProperties coolTeaProperties;

    private ThreadPoolExecutor workThreadPollExecutor;

    private final DelayContext delayContext;

    public DelayExecute(RedissonClient redissonClient, DelayContext delayContext) {
        this.redissonClient = redissonClient;
        this.delayContext = delayContext;

    }

    public void addDelayConfig(DelayQueue delayQueue, Method method, Object bean, DelayMethod delayMethod) {
        delayMethodMap.compute(delayQueue.name(), (k, v) -> {
            if (Objects.isNull(v)) {
                return Arrays.asList(createDelayMethodHandler(delayMethod));
            } else {
                v.add(createDelayMethodHandler(delayMethod));
                return v;
            }
        });
        queueNameList.add(delayQueue.name());
        RBlockingDeque<Message<?>> blockingDeque = redissonClient.getBlockingDeque(delayQueue.name());
        RDelayedQueue<Message<?>> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
        blockingQueueMap.put(delayQueue.name(), blockingDeque);
        delayedQueueMap.put(delayQueue.name(), delayedQueue);
    }

    private DelayMethodHandler createDelayMethodHandler(DelayMethod delayMethod) {
        return new DelayMethodHandler(delayMethod,
                dataSourceTransactionManager,
                transactionDefinition, redissonClient);
    }

    public void run(ApplicationReadyEvent applicationReadyEvent) {
        log.info("延迟队列组件开始运行。。。。。");
        ConfigurableApplicationContext applicationContext = applicationReadyEvent.getApplicationContext();
        this.dataSourceTransactionManager = applicationContext.getBean(DataSourceTransactionManager.class);
        this.transactionDefinition = applicationContext.getBean(TransactionDefinition.class);
        this.coolTeaProperties = applicationContext.getBean(CoolTeaProperties.class);
        ThreadPoolExecutor bossThreadPollExecutor = new ThreadPoolExecutor(coolTeaProperties.getDelayBoosThreadNum(), coolTeaProperties.getDelayBoosThreadNum(),
                5L, TimeUnit.MINUTES,
                new LinkedBlockingDeque<>(),
                ThreadFactoryBuilder.create()
                        .withDaemon(true)
                        .withNameFormat("delay-boos:%s")
                        .get()
        );
        workThreadPollExecutor = new ThreadPoolExecutor(coolTeaProperties.getDelayWorkThreadNum(), coolTeaProperties.getDelayWorkThreadNum(),
                5L, TimeUnit.MINUTES,
                new LinkedBlockingDeque<>(),
                ThreadFactoryBuilder.create()
                        .withDaemon(true)
                        .withNameFormat("delay-work:%s")
                        .get()
        );
        CoolTeaProperties coolTeaProperties = applicationContext.getBean(CoolTeaProperties.class);

        delayContext.doProcessBean();

        for (int i = 0; i < coolTeaProperties.getDelayBoosThreadNum(); i++) {
            bossThreadPollExecutor.execute(this::loop);
        }
    }

    private void loop() {
        for (; ; ) {
            blockingQueueMap.forEach((key, value) -> {
                Message<?> message = value.poll();
                if (Objects.nonNull(message)) {
                    executeDelay(message, key);
                }

            });
        }
    }

    void executeDelay(Message<?> message, String queueName) {
        CompletableFuture.runAsync(() -> {
            List<DelayMethodHandler> delayMethodList = delayMethodMap.get(queueName);
            delayMethodList.forEach(item -> item.handler(message));
        }, workThreadPollExecutor);
    }


    public void executeRunnable(Runnable r) {
        CompletableFuture.runAsync(r, workThreadPollExecutor);
    }
}
